package cn.zs.supply.common.core.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.io.UnsupportedEncodingException;


@Slf4j
public class RocketProducer {
    private static String topicPrefix = "";
    private static DefaultMQProducer producer;
    private static RocketProducer instance;
    private static TransactionMQProducer transactionMQProducer = null;

    private RocketProducer() {
    }

    /**
     * 获取RocketProducer实例
     *
     * @param addr
     * @param groupName
     * @param maxMessageSize
     * @param sendMsgTimeout
     * @return
     */
    public static RocketProducer getInstance(String msgtopicPrefix, String addr, String groupName, int maxMessageSize,
                                             int sendMsgTimeout, int sendMsgRetryTimes) {
        if (topicPrefix == null || topicPrefix.equals("")) {
            topicPrefix = msgtopicPrefix;
        }
        if (instance == null || producer == null) {
            synchronized (RocketProducer.class) {
                if (instance == null) {
                    instance = new RocketProducer();
                }
                if (producer == null) {
                    initProducer(addr, groupName, maxMessageSize, sendMsgTimeout, sendMsgRetryTimes);
                    //添加程序结束监听
                    Runtime.getRuntime().addShutdownHook(new ShutDownWork());
                }
            }
        }
        return instance;
    }

    /**
     * 关闭producer
     */
    public static void closeProducer() {
        if (producer != null) {
            producer.shutdown();
            producer = null;
        }
    }

    /**
     * 初始化producer
     *
     * @param addr
     * @param groupName
     * @param maxMessageSize
     * @param sendMsgTimeout
     */
    private static void initProducer(String addr, String groupName,
                                     int maxMessageSize,
                                     int sendMsgTimeout, int sendMsgRetryTimes) {
        /**
         * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
         * 注意：ProducerGroupName需要由应用来保证唯一<br>
         * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
         * 因为服务器会回查这个Group下的任意一个Producer
         */
        producer = new DefaultMQProducer(groupName);

        /**
         * Producer 参数说明
         *
         * producerGroup	DEFAULT_PRODUCER	Producer组名，多个Producer如果属于一个应用，发送同样的消息，则应该将他们归为同一组
         * createTopicKey	TBW102	支持在发送消息时，如果Topic不存在，自动创建Topic，但是要指定Key
         * topicdefaultTopicQueueNums	4	在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
         * sendMsgTimeout	10000	发送消息超时时间，单位毫秒
         * compressMsgBodyOverHowmuch	1024 * 4	消息Body超过多大开始压缩（Consumer收到消息会自动解压缩）,单位字节
         * retryTimesWhenSendFailed 2   发送失败后，重试几次
         * retryAnotherBrokerWhenNotStoreOK	FALSE	消息已经成功写入Master，但是刷盘超时或者同步到Slave失败，是否尝试尝试重试另一个Broker，不建议修改默认值，顺序消息无效
         * maxMessageSize	131072	客户端限制的消息大小，超过报错，同时服务端也会限制（默认128k)
         * transactionCheckListener	 	事务消息会查监听器，如果发送事务消息，必须设置
         * checkThreadPoolMinSize	1	Broker回查Producer事务状态时，线程池大小
         * checkThreadPoolMaxSize	1	Broker回查Producer事务状态时，线程池大小
         * checkRequestHoldMax	2000	Broker回查Producer事务状态时，Produceer本地缓冲请求队列大小
         */

        //设置发送的MQ地址
        producer.setNamesrvAddr(addr);

        //设置超时时间(3000毫秒)
        producer.setSendMsgTimeout(sendMsgTimeout);
        producer.setMaxMessageSize(maxMessageSize);
        //消息超出1024 * 4 = 4K进行压缩 客户端收到消息之前消息会自动解压
        producer.setCompressMsgBodyOverHowmuch(1024 * 4);

        //消息发送失败的重试次数
        producer.setRetryTimesWhenSendFailed(sendMsgRetryTimes);

        /**
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        try {
            producer.start();
        } catch (Exception ex) {
            log.error("初始化MQ异常：", ex);
        }
    }

    /**
     * 消息发送
     *
     * @param topic 主题
     * @param tags  tag
     * @param keys  键
     * @param msg   消息内容
     * @return
     * @throws UnsupportedEncodingException
     */
    public boolean sendMsg(String topic, String tags, String keys, String msg) throws UnsupportedEncodingException, UnsupportedEncodingException {
        Message message = new Message(topicPrefix + "_" + topic, tags, keys, msg.getBytes("UTF-8"));
        try {
            SendResult sendResult = producer.send(message);
            //log.debug("发送MQ状态：" + sendResult.getSendStatus());
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                return true;
            } else {
                log.error("消息发送失败:" + sendResult.toString());
                return false;
            }
        } catch (InterruptedException e) {
            log.error("消息发送失败" + e.getMessage(), e);
            // Restore interrupted state...
            Thread.currentThread().interrupt();
        } catch (RemotingException e) {
            log.error("消息发送失败" + e.getMessage(), e);
        } catch (MQClientException e) {
            log.error("消息发送失败" + e.getMessage(), e);
        } catch (MQBrokerException e) {
            log.error("消息发送失败" + e.getMessage(), e);
        }
        return false;
    }

    public boolean sendMsg(MessageQueueSelector selector, Object key, String topic, String tags, String keys, String msg) throws UnsupportedEncodingException, UnsupportedEncodingException {
        Message message = new Message(topicPrefix + "_" + topic, tags, keys, msg.getBytes("UTF-8"));
        try {
            SendResult sendResult = producer.send(message, selector, key);
            log.debug("发送MQ状态：" + sendResult.getSendStatus());
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                return true;
            }
        } catch (InterruptedException e) {
            log.error("消息发送失败" + e.getMessage(), e);
            // Restore interrupted state...
            Thread.currentThread().interrupt();
        } catch (RemotingException e) {
            log.error("消息发送失败" + e.getMessage(), e);
        } catch (MQClientException e) {
            log.error("消息发送失败" + e.getMessage(), e);
        } catch (MQBrokerException e) {
            log.error("消息发送失败" + e.getMessage(), e);
        }
        return false;
    }

}